package channel

import (
	"errors"
	"fmt"
	"micro-rulego/internal/biz/device"
	"sync"
	"time"

	"github.com/cenkalti/backoff/v4"
	"github.com/go-kratos/kratos/v2/log"
	"github.com/google/uuid"
)

type chanNode struct {
	ctrl     Channel
	entries  map[EntryID]ChannelEntry
	isActive bool
	exit     chan struct{}
}

type channelManagerUsecase struct {
	chans map[ChannelID]*chanNode
	connc device.DeviceManager
	m     sync.RWMutex
}

func (cm *channelManagerUsecase) ChannelNodeCreate(ctrl Channel) (ChannelID, error) {
	chanID := ChannelID(uuid.NewSHA1(uuid.NameSpaceURL, []byte(fmt.Sprintf("%s-%t", ctrl.Topic(), ctrl.Type()))).ID())
	cm.m.Lock()
	defer cm.m.Unlock()
	_, exist := cm.chans[chanID]
	if !exist {
		//if not exist, add Channel
		cm.chans[chanID] = &chanNode{
			ctrl:     ctrl,
			entries:  make(map[EntryID]ChannelEntry),
			isActive: false,
		}
	}
	return chanID, nil
}

func (cm *channelManagerUsecase) ChannelNodeDestroy(chanID ChannelID) error {
	cm.m.Lock()
	defer cm.m.Unlock()
	delete(cm.chans, chanID)
	return nil
}

func (cm *channelManagerUsecase) ChannelNodeAddEntry(chanID ChannelID, entry ChannelEntry) error {
	cm.m.Lock()
	defer cm.m.Unlock()
	node, exist := cm.chans[chanID]
	if !exist {
		return ErrNodeNotFound
	}
	if !node.ctrl.Type() && len(node.entries) != 0 {
		//a down Channel only has one connect
		return errors.New("Channel not support")
	}
	node.entries[entry.ID()] = entry
	return nil
}

func (cm *channelManagerUsecase) ChannelNodeDeleteEntry(chanID ChannelID, entry ChannelEntry) error {
	cm.m.Lock()
	defer cm.m.Unlock()
	node, exist := cm.chans[chanID]
	if !exist {
		return ErrNodeNotFound
	}
	delete(node.entries, entry.ID())
	return nil
}

func (cm *channelManagerUsecase) ChannelEnable(chanID ChannelID) error {
	cm.m.RLock()
	defer cm.m.RUnlock()
	node, exist := cm.chans[chanID]
	if !exist {
		return ErrNodeNotFound
	}

	if node.exit == nil {
		node.isActive = true
		node.exit = make(chan struct{})
	}

	if err := node.ctrl.Start(); err != nil {
		return err
	}

	if node.ctrl.Type() {
		// down
		go func() {
			for {
				select {
				case <-node.exit:
					return
				default:
					b, err := node.ctrl.Get()
					if err != nil && err != ErrChannelInvalid {
						log.Errorf("Channel node %d err", chanID)
						return
					} else if err == ErrChannelInvalid {
						log.Infof("Channel node %d disable", chanID)
						return
					}

					var wg sync.WaitGroup
					for _, v := range node.entries {
						wg.Add(1)
						go func(entry ChannelEntry) {
							defer wg.Done()

							// Transmit  backoff
							backoff := backoff.NewExponentialBackOff()
							backoff.InitialInterval = 100 * time.Millisecond
							backoff.MaxInterval = 5 * time.Second
							backoff.MaxElapsedTime = 30 * time.Second
							for {
								err := entry.Transmit(b)
								if err == nil {
									return
								}

								delay := backoff.NextBackOff()
								if delay == backoff.Stop {
									log.Errorf("Transmit failed after max retries: %v", err)
									return
								}

								log.Warnf("Transmit error: %v, retrying after %v", err, delay)
								time.Sleep(delay)
							}
						}(v)
					}
					wg.Wait()
				}
			}
		}()
	} else {
		// up
		for _, v := range node.entries {
			go func(entry ChannelEntry) {
				// use backoff
				backoff := backoff.NewExponentialBackOff()
				backoff.InitialInterval = 100 * time.Millisecond
				backoff.MaxInterval = 5 * time.Second
				backoff.MaxElapsedTime = 0 // never stop

				for {
					select {
					case <-node.exit:
						return
					default:
						b, err := entry.Receive()
						if err != nil {
							delay := backoff.NextBackOff()
							log.Warnf("Channel node %d receive err: %v, retrying after %v", chanID, err, delay)
							time.Sleep(delay)
							continue
						}
						// reset backoff
						backoff.Reset()

						if len(b) == 0 {
							log.Warnf("Channel node %d receive empty data", chanID)
							continue
						}

						if err := node.ctrl.Put(b); err != nil {
							log.Errorf("Channel node %d put err: %v", chanID, err)
							continue
						}
					}
				}
			}(v)
		}
	}
	return nil
}

func (cm *channelManagerUsecase) ChannelDisable(chanID ChannelID) error {
	cm.m.RLock()
	defer cm.m.RUnlock()
	node, exist := cm.chans[chanID]
	if !exist {
		return ErrNodeNotFound
	}
	if err := node.ctrl.Stop(); err != nil {
		return err
	}
	if node.exit != nil {
		close(node.exit)
		node.isActive = false
		node.exit = nil
	}
	return nil
}

func (cm *channelManagerUsecase) ChannelCheck() (map[ChannelID]*ChannelInfo, error) {
	cm.m.RLock()
	defer cm.m.RUnlock()

	result := make(map[ChannelID]*ChannelInfo)

	for channelID, node := range cm.chans {
		var status string
		if node.isActive {
			status = "active"
		} else {
			status = "inactive"
		}
		var typestr string
		if node.ctrl.Type() {
			typestr = "down"
		} else {
			typestr = "up"
		}
		entryInfos := make(map[EntryID][]uint32)
		for _, v := range node.entries {
			entryInfos[v.ID()] = append(entryInfos[v.ID()], v.DevID())
			entryInfos[v.ID()] = append(entryInfos[v.ID()], v.ConnID())
		}
		result[channelID] = &ChannelInfo{
			Status:     status,
			Type:       typestr,
			EntryInfos: entryInfos,
		}
	}

	return result, nil
}

func NewChannelManager(connc device.DeviceManager) ChannelManager {
	return &channelManagerUsecase{
		chans: make(map[ChannelID]*chanNode),
		connc: connc,
	}
}
